package com.loby.zk.zkeeper;

import com.loby.data.base.pojo.RtMonStationPollutant;
import com.loby.data.base.pojo.TMonMonitorStation;
import com.loby.monitor.client.propertie.ClientProperty;
import com.loby.zk.util.PropertyUtil;
import com.loby.zk.config.ZookeeperConfig;
import com.loby.data.base.pojo.TBaseOrganization;
import com.loby.data.base.pojo.TMonSourceShops;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * zookeeper操作工具
 */
@Component
@Slf4j
public class BaseZookeeper implements Watcher {

    public static final String MONITOR_SOURCES = "/MonitorSources";
    public static final String ONLINE_CLIENTS = "/OnLineClients";
    /***
     * 未注册的排放口，先存放在这里
     */
    public static final String CORP_PREFIX = "/Corp-";
    public static final String CORP_UN_REGS = "Corp-UnRegisters";
    public static final String SHOPS = "/SHOPs";
    public static final String SHOP_PREFIX="/Shop-";

    public static final String POLLUTION_PRODUCTION_FACILITIES = "/PPFs";
    public static final String POLLUTION_PRODUCTION_FACILITIES_PREFIX = "/PPF-";

    public static final String POLLUTION_CONTROL_FACILITIES = "/PCFs";
    public static final String POLLUTION_CONTROL_FACILITIES_PREFIX = "/PCF-";

    public static final String CLIENT_NAME = "/name";
    public static final String CORPORATION_NAME = "/name";
    public static final String SHOP_NAME = "/name";
    public static final String MONITOR_STATION_NAME = "/name";
    public static final String CLIENT_PROTOCOL = "/Protocol";
    public static final String CLIENT_INFO_TOPIC = "ClientInfo";
    public static final String CLIENT_CODE = "/ClientCode";
    public static final String NEED_RESPONSE = "/NeedResponse";
    public static final String PWD = "/PWD";
    public static final String TABLE_NAME_PREFIX = "/his_";
    public static final String REAL_TABLE_NAME_SUFFIX = "_real_data";
    public static final String MINUTE_TABLE_NAME_SUFFIX = "_minute_data";
    public static final String HOUR_TABLE_NAME_SUFFIX = "_hour_data";
    public static final String DAY_TABLE_NAME_SUFFIX = "_day_data";
    public static final String EXCESS_TABLE_NAME_SUFFIX = "_excess_data";
    public static final String CLIENT_PROPERTIES_PROPERTYS = "/ClientProperties";


    public static final String PRO_REAL_TABLE_NAME = "/RealTableName";
    public static final String PRO_MIN_TABLE_NAME = "/MinTableName";
    public static final String PRO_HOUR_TABLE_NAME = "/HourTableName";
    public static final String PRO_DAY_TABLE_NAME = "/DayTableName";
    public static final String PRO_EXCESS_TABLE_NAME = "/ExcessTableName";




    public static final String UN_KNOW_CLIENT_TAG = "UnKnowClient";
    public static final String UN_KNOW_CORPORATION_NAME_PREFIX = "UK_";
    public static final String UN_KNOW_WORKSHOP_NAME = "Undefined Workshop";
    public static final int ZK_MIN_INFO_PATH_LEN = 3;
    public static final int ZK_ORG_INFO_PATH_LEN = 3;
    public static final int ZK_SHOP_INFO_PATH_LEN = 5;
    public static final int ZK_BASE_INFO_PATH_LEN = 7;
    public String OFF_LINE_TIME = "/OffLineTime";
    public static final String CONDITIONS = "/Conditions";
    private ZooKeeper zookeeper;

    @Autowired
    private ZookeeperConfig config;

    public BaseZookeeper() {
        System.out.println("注入");

    }


    /**
     *      * 超时时间
     *     
     */

    private static final int SESSION_TIME_OUT = 30000;

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    @Override
    public void process(WatchedEvent event) {
        if (event.getState() == KeeperState.SyncConnected) {
            System.out.println("Watch received event");
            countDownLatch.countDown();
        }
    }

    /**
     * 初始化zookeeper，建立连接，创建基础路径
     */
    public boolean checkZookeeper()
    {
            try {
                connectZookeeper();
                if (!exists(ONLINE_CLIENTS, false)) {
                    createNode(ONLINE_CLIENTS);
                }
                if (!exists(MONITOR_SOURCES, false)) {
                    createNode(ONLINE_CLIENTS);
                }
                return true;
            } catch (KeeperException | InterruptedException  e) {
                log.error("zookeeper初始化失败:",e);
            } catch (Exception e) {
                log.error("zookeeper初始化失败:",e);
            }
            return false;
    }

    public boolean checkZookeeper(String path)
    {
        try {
                 connectZookeeper();
                if (!exists(path, false)) {
                    createNode(path);
                }

            return true;
        } catch (KeeperException | InterruptedException  e) {
            log.error("创建"+path+"失败",e);
        } catch (Exception e) {
            log.error("创建"+path+"失败",e);
        }
        return false;
    }

    /**
     * 连接zookeeper
     *     * @param host
     *     * @throws Exception
     *    
     */
    public void connectZookeeper() throws Exception
    {
        if(zookeeper == null || !zookeeper.getState().isAlive() || !zookeeper.getState().isConnected())
        {
            closeConnection();
            zookeeper = new ZooKeeper(config.getHosts(), SESSION_TIME_OUT, this);
            countDownLatch.await();
        }
        log.info("zookeeper connection success[" + config.getHosts() + "]");
    }


    public void connectZookeeper(String host) throws Exception {
        zookeeper = new ZooKeeper(host, SESSION_TIME_OUT, this);
        countDownLatch.await();
        System.out.println("zookeeper connection success");
    }

    /**
     * 创建节点
     *
     * @param path
     * @throws Exception
     */
    public String createNode(String path) throws Exception {
        return this.zookeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     *     * 创建节点
     *     * @param path
     *     * @param data
     *     * @throws Exception
     *    
     */
    public String createNode(String path, String data) throws Exception {
        return this.zookeeper.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     *     * 获取路径下所有子节点
     *     * @param path
     *     * @return
     *     * @throws KeeperException
     *     * @throws InterruptedException
     *    
     */
    public List<String> getChildren(String path) throws KeeperException, InterruptedException {
        List<String> children = zookeeper.getChildren(path, false);
        return children;
    }

    public boolean exists(String path, boolean watch) throws KeeperException, InterruptedException {
        return null!=zookeeper.exists(path, watch);
    }

    public boolean exists(String path, Watcher watch) throws KeeperException, InterruptedException {
        return null!=zookeeper.exists(path, watch);
    }

    /**
     *     * 获取节点上面的数据
     *     * @param path  路径
     *     * @return
     *     * @throws KeeperException
     *     * @throws InterruptedException
     *    
     */
    public String getData(String path) throws KeeperException, InterruptedException {
        byte[] data = zookeeper.getData(path, false, null);
        if (data == null) {
            return "";
        }
        return new String(data);
    }

    /**
     *     * 设置节点信息
     *     * @param path  路径
     *     * @param data  数据
     *     * @return
     *     * @throws KeeperException
     *     * @throws InterruptedException
     *    
     */


    public Stat setData(String path, String data) throws KeeperException, InterruptedException {
        Stat stat = zookeeper.setData(path, data.getBytes(), -1);
        return stat;
    }

    /**
     *     * 删除节点
     *     * @param path
     *     * @throws InterruptedException
     *     * @throws KeeperException
     *    
     */


    public void deleteNode(String path) throws InterruptedException, KeeperException {
        zookeeper.delete(path, -1);


    }


  /*  public void deleteSubNode(String nodeStr) throws KeeperException, InterruptedException {
        String nodePath = nodeStr;
        //打印当前节点路径
        System.out.println("Node Path >>>>>>>>> [" + nodePath + " ]");

        if (zookeeper.getChildren(nodePath, false).size() == 0) {
            //删除节点
            System.out.println("Deleting Node Path >>>>>>>>> [" + nodePath + " ]");
              zookeeper.delete(nodePath,-1);
        } else {
            //递归查找非空子节点
            List<String> list = zookeeper.getChildren(nodeStr, true);
            System.out.println(list);
            for (String str : list) {
                deleteSubNode(nodePath + "/" + str);
            }
        }
    }*/

    /**
     *     * 获取创建时间
     *     * @param path
     *     * @return
     *     * @throws KeeperException
     *     * @throws InterruptedException
     *    
     */


    public String getCTime(String path) throws KeeperException, InterruptedException {
        Stat stat = zookeeper.exists(path, false);
        return String.valueOf(stat.getCtime());
    }

    /**
     *     * 获取某个路径下孩子的数量
     *     * @param path
     *     * @return
     *     * @throws KeeperException
     *     * @throws InterruptedException
     *    
     */


    public Integer getChildrenNum(String path) throws KeeperException, InterruptedException {
        int childenNum = zookeeper.getChildren(path, false).size();
        return childenNum;
    }


    /**
     *     * 关闭连接
     *     * @throws InterruptedException
     *    
     */
    public void closeConnection() throws InterruptedException {
        if (zookeeper!= null) {
            zookeeper.close();
        }
    }


    public boolean newOrganization(TBaseOrganization organization) {
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+ organization.getPkOrgCode();
        if(checkZookeeper())
        {
            try {
                createNode(path, organization.getOrgName());
                return true;

            } catch (Exception e) {
                log.error("插入数据到zookeeper失败！",e);
            }
        }
        return false;
    }

    public boolean removeOrganization(TBaseOrganization baseOrganization){
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+baseOrganization.getPkOrgCode();
        try {
            if(checkZookeeper()){
                ZKUtil.deleteRecursive(zookeeper,path);
                return true;
            }
        } catch (Exception e) {
            log.error("删除失败",e);
        }
        return false;
    }

    public boolean updateOrganizationName(TBaseOrganization organization){
    String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode();
        if(checkZookeeper()){
                try {
                    setData(path,organization.getOrgName());
                } catch (Exception e) {
                    log.error("修改失败",e);
                    return false;
                }
        }
            return true;
    }

   public boolean newShop(TBaseOrganization organization, TMonSourceShops shop){
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode();
        if (checkZookeeper()){
            try {
                createNode(path+SHOPS);
                createNode(path+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo(),shop.getShopName());
                createNode(path+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo()+POLLUTION_CONTROL_FACILITIES,"污染源治理设施");
                return true;
            } catch (Exception e) {
                log.error("插入车间数据失败",e);
            }

        }
        return false;
   }

    public boolean removeShop(TBaseOrganization organization, TMonSourceShops shop){
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode()+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo();
        try {
            if(checkZookeeper()){
                ZKUtil.deleteRecursive(zookeeper,path);
                return true;
            }
        } catch (Exception e) {
           log.error("删除失败",e);
        }
        return false;
    }

    public boolean updateShop(TMonSourceShops shop, TBaseOrganization organization) {
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode()+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo();
        try {
            if(checkZookeeper()){
                setData(path,shop.getShopName());
                return true;
            }
        } catch (Exception e) {
           log.info("修改车间失败",e);
        }
        return false;
    }

    public boolean newMonitorStation(TMonMonitorStation station,TMonSourceShops shop,TBaseOrganization organization,String json){
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode()+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo()+POLLUTION_CONTROL_FACILITIES+"/"+station.getStationUniqueCode();
        try {
            if(checkZookeeper(path)){
                if (!exists(path, false)) {
                    createNode(path,json);
                    return true;
                }else {
                    setData(path,json);
                    return true;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean removeMonitorStation(TMonMonitorStation station,TMonSourceShops shop,TBaseOrganization organization){
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode()+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo()+POLLUTION_CONTROL_FACILITIES+"/"+station.getStationUniqueCode();
        try {
            if(checkZookeeper()){
                ZKUtil.deleteRecursive(zookeeper,path);
                return true;
            }
        } catch (Exception e) {
            log.error("删除失败",e);
        }
        return false;

    }

    public boolean updateByMonitorStationId(TBaseOrganization organization,TMonSourceShops shop,TMonMonitorStation station,String json){
        String path=BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode()+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo()+POLLUTION_CONTROL_FACILITIES+"/"+station.getStationUniqueCode();


           if(checkZookeeper()) {
               try {
                   setData(path,json);
                   return true;
               } catch (Exception e) {
                   log.error("删除失败",e);
               }
           }
           return false;

    }


 /*   public String findDataByPath(String path) {
        try {

            return getData(path);

        } catch (Exception e) {
            log.info(e.getLocalizedMessage());
        }
        return null;
    }*/

    public boolean addExcess(TBaseOrganization organization, TMonSourceShops shop, TMonMonitorStation station, List<RtMonStationPollutant> pollutants) {
        String path= BaseZookeeper.MONITOR_SOURCES+BaseZookeeper.CORP_PREFIX+organization.getPkOrgCode()+SHOPS+"/"+organization.getPkOrgCode()+":"+shop.getShopNo()+POLLUTION_CONTROL_FACILITIES+"/"+station.getStationUniqueCode();

        try {
            if(checkZookeeper()) {
                if(exists(path,false)){
                    String data = getData(path);
                    ClientProperty property = ClientProperty.fromJson(data);
                    ClientProperty property1 = PropertyUtil.addCondition(property, pollutants);
                    setData(path,property1.toJosn());
                }else {
                   log.info("排放口不存在！");
                   return false;
               }
            }
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
}

